package com.yuan.mqtt.client;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

/**
 * @Classname MqttClient
 * @Description :
 * @Date 2021/9/14 15:25
 * @Author cfy
 */
@Component
public class MqttPushClient implements ApplicationRunner {
    private static MqttClient client;

    /**
     * MQTT连接属性配置对象--
     */
    public final MqttConfig mqttConfig;

    public MqttPushClient(MqttConfig mqttConfig) {
        this.mqttConfig = mqttConfig;
    }

    /**
     * 初始化参数配置
     * 项目启动后加载，防止数据初始化失败
     */
    @Override
    public void run(ApplicationArguments args) throws Exception {
        this.connect();
    }


    /**
     * 用来连接服务器
     */
    private void connect() throws Exception {
        client = new MqttClient(mqttConfig.getHost(), mqttConfig.getClientid(), new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(false);
        options.setUserName(mqttConfig.getUsername());
        options.setPassword(mqttConfig.getPassword().toCharArray());
        options.setCleanSession(false);   //是否清除session
        // 设置超时时间
        options.setConnectionTimeout(30);
        // 设置会话心跳时间
        options.setKeepAliveInterval(20);
        try {
            String[] msgtopic = mqttConfig.getMsgTopic();
            //订阅消息
            int[] qos = new int[msgtopic.length];
            for (int i = 0; i < msgtopic.length; i++) {
                qos[i] = 2;
            }
            client.setCallback(new PushCallback(client, options, msgtopic, qos));
            client.connect(options);
            client.subscribe(msgtopic, qos);
            System.out.println("MQTT连接成功:" + mqttConfig.getClientid() + ":" + client);
        } catch (Exception e) {
            System.out.println("MQTT连接异常：" + e);
        }
    }

    /**
     * 发布，默认qos为2，持久化
     * @param topic
     * @param pushMessage
     */

    public void pushlish(String topic, String pushMessage) {
        pushlish(2, false, topic, pushMessage);
    }

    /**
     * 发布
     * @param qos         连接方式
     * @param retained    是否保留
     * @param topic       主题
     * @param pushMessage 消息体
     */

    public void pushlish(int qos, boolean retained, String topic, String pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        MqttTopic mqttTopic = client.getTopic(topic);
        if (null == mqttTopic) {
            System.out.println("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            token = mqttTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        }

    }


    /**
     * 订阅某个主题
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos) {
        try {
            System.out.println("topic:" + topic);
            client.subscribe(topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

}
